package org.ray.pattern.jobmanager.single;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import org.ray.util.Pause;

public abstract class AbstractJobManagerImpl<T> implements JobManager<T>
{
	protected Map<T, JobInfo<T>> registeredMemoryMap = 
		new HashMap<T, JobInfo<T>>();

	protected Timer memoryTimeoutCheckTimer;
	
	private long jobTimeout = 14400L;
	private long checkInterval = 15L;
	private int workerCount = 5;
	private long waitUnlockTick = 15;
	private long waitUnlockTickCount = 8;
	
	
	public AbstractJobManagerImpl()
	{	    
	    // start a thread pool to send alert email
	    System.out.println("Start " + workerCount + "worker threads");
	    
	    memoryTimeoutCheckTimer = new Timer();
	    
	    memoryTimeoutCheckTimer.schedule(new TimerTask()
	    {
			public void run()
			{
				// TODO Auto-generated method stub
				Collection<JobInfo<T>> values = null;
				synchronized(registeredMemoryMap)
				{
					values = new ArrayList<JobInfo<T>>(registeredMemoryMap.values());
				}
				
				JobInfo<T> jobInfoExisting = null;
				Iterator<JobInfo<T>> it = values.iterator();
				
				// Scan to see if there is memory compilation has reached the timeout
				while (it.hasNext())
				{
					jobInfoExisting = it.next();
					if (jobInfoExisting != null)
					{
						synchronized(jobInfoExisting)
						{
							if (!jobInfoExisting.isAlerted() &&
									isJobCancellable(jobInfoExisting.getKey()))
							{
								// only alert once
								jobInfoExisting.setAlerted(true);

								// send email to user
								alertTimeout(jobInfoExisting);
							}
						}
					}
				}
				
			}
	    	
	    }
	    , checkInterval, checkInterval);
	}

	public JobInfo<T> tryLockJob(JobInfo<T> jobInfo)
	{
		// TODO Auto-generated method stub
		T key = jobInfo.getKey();
		// TODO Auto-generated method stub
		JobInfo<T> jobInfoExisting = null;
		synchronized (registeredMemoryMap)
		{
			jobInfoExisting = registeredMemoryMap.get(key);

			if (jobInfoExisting == null)
			{
				registeredMemoryMap.put(key, jobInfo.copy());
				return null;
			}
		}
		if (isJobCancellable(key))
		{
			// submit cancel memory first
			cancelJob(key);

			// wait memory to be unlocked for a period 
			// If failed, force to unlock it.
			if (!waitJobUnlocked(key))
			{
				forceUnlockJob(key);
			}
			
			// Register new job
			registeredMemoryMap.put(key, jobInfo.copy());
			return null;
		}
		else
		{
			// If still compiling, check sungrid id
			if (jobInfoExisting.getState() == JobInfo.JobState.STATE_RUNNING)
			{
				doBeforeReturnLock(jobInfoExisting);
				
				return jobInfoExisting.copy();
			}
			else
			{
				// wait memory to be unlocked for a period 
				// If failed, force to unlock it.
				if (!waitJobUnlocked(key))
				{
					forceUnlockJob(key);
				}
				
				// Register new memory compilation
				registeredMemoryMap.put(key, jobInfo.copy());

				return null;
			}
		}
	}

	protected boolean forceUnlockJob(T key)
	{
		System.out.println("Warning: Force cancel job: " + key);
		boolean ret = false;
		synchronized (registeredMemoryMap)
		{
			ret = (registeredMemoryMap.remove(key) != null);
		}
		
		return ret;
	}

	private boolean waitJobUnlocked(T key)
	{
		// Wait memory to be unlocked
		JobInfo<T> jobInfoExisting = null;
        int counter = 0;
		do
		{
			Pause.manySeconds(this.waitUnlockTick);
			counter ++;
			synchronized (registeredMemoryMap)
			{
				jobInfoExisting = registeredMemoryMap.get(key);
			}
		}
		while(jobInfoExisting != null && counter < this.waitUnlockTickCount);
		
		return (jobInfoExisting == null);
	}

	public boolean unlockJob(T key, long timestamp)
	{
		// TODO Auto-generated method stub
		JobInfo<T> jobInfoExisting = null;
		boolean ret = false;
		synchronized (registeredMemoryMap)
		{
			jobInfoExisting = registeredMemoryMap.get(key);
			// if it is not force unlocked
			if (jobInfoExisting != null && 
					jobInfoExisting.getStartTimestamp() == timestamp)
			{
				ret = (registeredMemoryMap.remove(key) != null);
			}
		}
		
		return ret;
	}

	public void updateJob(JobInfo<T> jobInfo)
	{
		// TODO Auto-generated method stub
		JobInfo<T> jobInfoExisting = null;
		synchronized (registeredMemoryMap)
		{
			jobInfoExisting = registeredMemoryMap.get(jobInfo.getKey());
		}

		if (jobInfoExisting != null)
		{
			synchronized (jobInfoExisting)
			{
				jobInfoExisting.update(jobInfo);
			}
		}
	}

	public void cancelJob(T key)
	{
		// TODO Auto-generated method stub
		JobInfo<T> jobInfo = null;
		synchronized (registeredMemoryMap)
		{
			jobInfo = registeredMemoryMap.get(key);
		}

		if (jobInfo != null)
		{
			synchronized (jobInfo)
			{
				if (jobInfo.getState() == JobInfo.JobState.STATE_FINISHED)
				{
					System.out.println("Fail to Cancel because memory compilation finished:" + key);
					return;
				}

				jobInfo.setState(JobInfo.JobState.STATE_CANCELLED);
				
				doCancelJob(jobInfo);
			}
		}
		else
		{
			System.out.println("Fail to Cancel because job compilation finished:" + key);
		}
	}

	public JobInfo<T> getRunningJob(T key)
	{
		// TODO Auto-generated method stub
		JobInfo<T> jobInfo = null;
		JobInfo<T> jobInfoRet = null;

		synchronized (registeredMemoryMap)
		{
			jobInfo = registeredMemoryMap.get(key);
			if (jobInfo != null)
			{
				synchronized (jobInfo)
				{
					if (jobInfo.getState() == JobInfo.JobState.STATE_RUNNING)
						jobInfoRet = jobInfo.copy();
				}
			}
		}
		
		return jobInfoRet;
	}

	public List<JobInfo<T>> getRunningJobList()
	{
		// TODO Auto-generated method stub
		Collection<JobInfo<T>> values = null;

		synchronized(registeredMemoryMap)
		{
			System.out.println("Is Empty:" + registeredMemoryMap.isEmpty());
			values = new ArrayList<JobInfo<T>>(registeredMemoryMap.values());
		}
		
		List<JobInfo<T>> newList = new LinkedList<JobInfo<T>>();
		JobInfo<T> jobInfoExisting = null;
		
		Iterator<JobInfo<T>> it = values.iterator();
		while (it.hasNext())
		{
			jobInfoExisting = it.next();
			if (jobInfoExisting != null)
			{
				synchronized(jobInfoExisting)
				{
					if (jobInfoExisting.getState() == JobInfo.JobState.STATE_RUNNING)
					{
						newList.add(jobInfoExisting.copy());
					}
				}
			}
		}
		
		return newList;
	}

	public boolean isJobCancellable(T key) {
		// TODO Auto-generated method stub
		JobInfo<T> jobInfo = null;
		
		boolean isCancellable = true;
		synchronized (registeredMemoryMap)
		{
			jobInfo = registeredMemoryMap.get(key);
			if (jobInfo != null)
			{
				synchronized (jobInfo)
				{
					long lTimeExpired = 
						(System.currentTimeMillis() - jobInfo.getStartTimestamp()) / 1000;
					if (lTimeExpired < jobTimeout)
					{
						isCancellable = false;
					}
					else
					{
						// if canceled or compiled, no need to cancel
						if (jobInfo.getState() != JobInfo.JobState.STATE_RUNNING)
						{
							System.out.println("Warning:Job timeout but cancelled or compiled:" + key);
							isCancellable = false;
						}
					}
				}
			}
		}
		return isCancellable;
	}

	public boolean isJobCancelled(T key)
	{
		// TODO Auto-generated method stub
		if (key == null)
			return false;

		JobInfo<T> jobInfo = null;
		synchronized (registeredMemoryMap)
		{
			jobInfo = registeredMemoryMap.get(key);
		}

		boolean ret = false;
		if (jobInfo != null)
		{
			synchronized (jobInfo)
			{
				ret = (jobInfo.getState() == JobInfo.JobState.STATE_CANCELLED);
			}
		}

		return ret;
	}

	public void setJobFinished(T key)
	{
		// TODO Auto-generated method stub
		JobInfo<T> jobInfo = null;
		synchronized (registeredMemoryMap)
		{
			jobInfo = registeredMemoryMap.get(key);
		}

		if (jobInfo != null)
		{
			synchronized (jobInfo)
			{
				jobInfo.setState(JobInfo.JobState.STATE_FINISHED);
			}
		}
	}

	public long getJobRunningTimeout()
	{
		// TODO Auto-generated method stub
		return jobTimeout;
	}

	public void setJobRunningTimeout(long jobTimeout)
	{
		// TODO Auto-generated method stub
		this.jobTimeout = jobTimeout;
	}

	public int getWorkerCount()
	{
		// TODO Auto-generated method stub
		return this.workerCount;
	}

	public void setWorkerCount(int workerCount)
	{
		// TODO Auto-generated method stub
		this.workerCount = workerCount;
	}

	public long getWaitUnlockTick()
	{
		// TODO Auto-generated method stub
		return this.waitUnlockTick;
	}
	public void setWaitUnlockTick(long tick)
	{
		// TODO Auto-generated method stub
		this.waitUnlockTick = tick;
	}

	public long getWaitUnlockTickCount()
	{
		// TODO Auto-generated method stub
		return this.waitUnlockTickCount;
	}
	public void setWaitUnlockTickCount(long tickCount)
	{
		// TODO Auto-generated method stub
		this.waitUnlockTickCount = tickCount;
	}

	public long getJobCheckInterval()
	{
		// TODO Auto-generated method stub
		return this.checkInterval;
	}

	public void setJobCheckInterval(long interval)
	{
		// TODO Auto-generated method stub
		this.checkInterval = interval;
	}
	
	protected abstract void doBeforeReturnLock(JobInfo<T> jobInfo);
	
	protected abstract void doCancelJob(JobInfo<T> jobInfo);
	
	protected abstract void alertTimeout(JobInfo<T> jobInfo);

}
